ea2dd50805dc96e8b322c80e7b1633006d472c4c,src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java,KafkaMessageDrivenChannelAdapter,toMessage,#Object#Object#KafkaMessageMetadata#Acknowledgment#,196
Before Change
.withKeyValue(KafkaHeaders.MESSAGE_KEY, key)
.withKeyValue(KafkaHeaders.TOPIC, metadata.getPartition().getTopic())
.withKeyValue(KafkaHeaders.PARTITION_ID, metadata.getPartition().getId())
.withKeyValue(KafkaHeaders.OFFSET, metadata.getOffset())
.withKeyValue(KafkaHeaders.NEXT_OFFSET, metadata.getNextOffset());
// pre-set the message id header if set to not generate
if (!this.generateMessageId) {
After Change
headers.put(KafkaHeaders.MESSAGE_KEY, key);
headers.put(KafkaHeaders.TOPIC, metadata.getPartition().getTopic());
headers.put(KafkaHeaders.PARTITION_ID, metadata.getPartition().getId());
headers.put(KafkaHeaders.OFFSET, metadata.getOffset());
headers.put(KafkaHeaders.NEXT_OFFSET, metadata.getNextOffset());
// pre-set the message id header if set to not generate